package spark_core.operate_transform;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

/**
 * @author shihb
 * @date 2020/1/8 18:34
 * 单元训练，综合运用算子
 * 需求，统计每个省份广告被点击次数的topN
 * 数据结构 userId,adId,province,city,timestamp
 */
public class UnitTraining {

  public static void main(String[] args) {


    //local模式,创建SparkConf对象设定spark的部署环境
    SparkConf sparkConf = new SparkConf().setMaster("local[*]").setAppName("ad click topN");
    //创建spark上下文对象（这边是java上下文）
    JavaSparkContext sc = new JavaSparkContext(sparkConf);
    JavaRDD<String> inputRdd = sc
        .textFile("D:\\SHBData\\IDEAProjects\\spark-parent\\SparkTestDemo\\src\\main\\resources\\AdClickLog.csv");
    JavaPairRDD<Tuple2<String, String>, Integer> aggResultRdd = inputRdd
        .mapToPair(s -> {
          String[] arr = s.split(",");
//      long userId = Long.parseLong(arr[0].trim());
//      long adId = Long.parseLong(arr[1].trim());
//      String province = arr[2].trim();
//      String city = arr[3].trim();
//      long timestamp = Long.parseLong(arr[4].trim());
//      return new AdClickEvent(userId, adId, province, city, timestamp);
          return new Tuple2<Tuple2<String, String>, Integer>(
              new Tuple2<>(arr[2].trim(), arr[1].trim()), 1);
        }).reduceByKey((o1, o2) -> o1 + o2);

    JavaPairRDD<String, Iterable<Tuple2<String, Integer>>> groupRdd = aggResultRdd
        .mapToPair(
            t -> new Tuple2<String, Tuple2<String, Integer>>(t._1._1, new Tuple2<>(t._1._2, t._2)))
        .groupByKey();
    JavaPairRDD<String, List<Tuple2<String, Integer>>> result = groupRdd
        .mapValues(iter -> {
          ArrayList<Tuple2<String, Integer>> list = new ArrayList<>();
          Iterator<Tuple2<String, Integer>> iterator = iter.iterator();
          while (iterator.hasNext()) {
            list.add(iterator.next());
          }
          list.sort((t1, t2) -> t2._2 - t1._2);
          return new ArrayList<>(list.subList(0,Math.min(3,list.size())));
        });
    result.collect().forEach(System.out::println);




  }
}
